package study.cdc.debezium.support.debezium;

import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.Configuration;

@Slf4j
@Configuration
public class DebeziumRunner implements CommandLineRunner {

    @Autowired
    private io.debezium.config.Configuration config;

    @Override
    public void run(String... args) throws Exception {
        DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class)
                .using(config.asProperties())
//                .using((success, message, error) -> {
//                    log.info(">>>>>>>>>>>>> cal");
//                    // 强烈建议加上此部分的回调代码，方便查看错误信息
//                    if (!success && error != null) {
//                        // 报错回调
//                        log.error(message);
//                        log.error("", error);
//                    }
//                })
                //消费数据
                .notifying(new ChangeEventConsumer()::handle)
                //                .notifying(new ChangeEventHandler()::handleBatch)
                .build();

        Thread t = new Thread(engine, "Debezium-Thread");
        t.start();
    }
}
